package SparkStreaming;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;
import scala.collection.Iterator;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

public class HiveSourceInput implements Serializable{


    /**
     * Use Hive Database
     */
    private String hiveDbName;

    /**
     * Select SQL
     */
    private  String sqlText;

    /**
     * 只生成一个 HiveContext 的实例，否则可能内存溢出
     */
    protected static HiveContext hiveContext = null;

    private transient JavaSparkContext sc;

    protected static Log LOG = LogFactory.getLog(HiveSourceInput.class);

    public void init(){
        sc = new JavaSparkContext("local","HiveSourceInput");
        hiveContext = new HiveContext(sc.sc());

    }

    public HiveSourceInput() {
        init();
    }
    public HiveSourceInput(JavaSparkContext sc,HiveContext hiveContext,String hiveDbName,String sqlText){
        this.sc = sc;
        this.hiveContext = hiveContext;
        this.hiveDbName = hiveDbName;
        this.sqlText = sqlText;
    }

    public JavaRDD<String> getSourceRDD() {
        if(hiveContext == null) {
            hiveContext = new HiveContext(sc.sc());
        } else {
            //Sparkcontext 和 Hive 中持有的 SparkContext 不是一个对象，则说明 SaprkContext 被重启过
            if(hiveContext.sparkContext() != sc.sc()) {
                hiveContext = null;
                hiveContext = new HiveContext(sc.sc());
            }
        }
        hiveContext.sql("use "+this.hiveDbName);
        //List<Row> tables = hiveContext.sql("show tables").collectAsList();
        //LOG.info("show tables: " + tables);

        DataFrame dataFrame = hiveContext.sql(this.sqlText);
        //sql.show();

        return dataFrame.toJavaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                //FlowData data = new FlowData();
                /**
                 * 获取表的字段名称
                 */
                StructType schema = row.schema();
                String[] fieldNames = schema.fieldNames();

                JSONObject jsonObject = new JSONObject();
                int length = row.length();
                for (int i = 0; i < length; i++) {
                    Object o = row.get(i);

                    if(o == null) {
                        jsonObject.put(fieldNames[i], null);
                    } else if(o instanceof scala.collection.mutable.ArrayBuffer) {
                        //处理 数组 类型
                        scala.collection.mutable.ArrayBuffer<String> oldData = (scala.collection.mutable.ArrayBuffer<String>)o;
                        ArrayList<String> listData = new ArrayList<String>();
                        if(!oldData.isEmpty()){
                            Iterator<String> iterator = oldData.iterator();
                            while(iterator.hasNext()){
                                String next = iterator.next();
                                listData.add(next);
                            }
                        }
                        jsonObject.put(fieldNames[i], listData);
                    } else if(o instanceof scala.collection.immutable.List) {
                        //处理 list 类型
                        scala.collection.immutable.List<String> oldData = (scala.collection.immutable.List<String>)o;
                        ArrayList<String> listData = new ArrayList<String>();
                        if(!oldData.isEmpty()){
                            Iterator<String> iterator = oldData.iterator();
                            while(iterator.hasNext()){
                                String next = iterator.next();
                                listData.add(next);
                            }
                        }
                        jsonObject.put(fieldNames[i], listData);
                    } else if(o instanceof GenericRowWithSchema) {
                        // 处理 Map 类型
                        GenericRowWithSchema oldData = (GenericRowWithSchema)o;
                        Map<String, Object> mapData = new HashMap<String, Object>();

                        StructType  innerSchema = oldData.schema();

                        Object[] values = oldData.values();
                        String[] names = innerSchema.fieldNames();
                        for (int j = 0; j < names.length; j++) {
                            mapData.put(names[j], values[j]);
                        }

                        jsonObject.put(fieldNames[i], mapData);
                    } else if(o instanceof scala.collection.immutable.Map) {
                        // 处理 Map 类型
                        scala.collection.immutable.Map oldData = (scala.collection.immutable.Map)o;
                        Map<String, Object> mapData = new HashMap<String, Object>();

                        final Iterator iterator = oldData.toIterator();
                        while (iterator.hasNext()) {
                            final Tuple2<String, Object> tuple = (Tuple2<String, Object>) iterator.next();
                            mapData.put(tuple._1(), tuple._2());
                        }

                        jsonObject.put(fieldNames[i], mapData);
                    } else {
                        //处理简单类型
                        jsonObject.put(fieldNames[i], o);
                    }
                }
                return jsonObject.toJSONString();
            }
        });
    }


    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[1]").setAppName("HiveTest");
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        HiveContext hiveContext = new HiveContext(sparkContext.sc());
        String hiveDbName = "etl";
        //select scores from rawtravelscenicinfo where  projectId='8' and batchId='4028e6875551bcab015552c6d24c0005' and  size(scores)>0;
        String sqlString = "select  *  from rawtravelscenicinfo where projectid='8' and batchid='d517b20654ad4733a304061a49c89868' limit 10";

        HiveSourceInput hiveSourceInput = new HiveSourceInput(sparkContext,hiveContext,hiveDbName,sqlString);
        JavaRDD<String> sourceRDD = hiveSourceInput.getSourceRDD();

        sourceRDD.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });

    }


}
